package com.java.code.kafka;

import org.apache.kafka.clients.consumer.Consumer;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.common.TopicPartition;
import org.apache.kafka.common.utils.Bytes;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.kafka.core.DefaultKafkaConsumerFactory;
import org.springframework.kafka.listener.AcknowledgingMessageListener;
import org.springframework.kafka.listener.ConcurrentMessageListenerContainer;
import org.springframework.kafka.listener.ContainerProperties;
import org.springframework.kafka.listener.KafkaMessageListenerContainer;
import org.springframework.kafka.support.Acknowledgment;
import org.springframework.kafka.support.TopicPartitionInitialOffset;
import org.springframework.kafka.support.TopicPartitionOffset;
import org.springframework.util.SerializationUtils;

import java.util.*;
import java.util.concurrent.ConcurrentHashMap;
import java.util.regex.Pattern;

/**
 * @author weichuanyuan
 * @e-mail wei.cy@bsoft.com.cn
 * @date 2021/10/19 21:42
 * @package com.java.code.kafka
 * @description
 */
public class MQConsumerByKafka implements MQConsumer, AcknowledgingMessageListener<String, Bytes> {
    private Properties properties;
    private Map<String, Integer> mapTopic2Concurrency = new ConcurrentHashMap();
    private Map<String, MQBusinessHandler> mapTopic2Handler = new ConcurrentHashMap();
    private List<MQConsumerInfo> listTopic2Handler = new ArrayList();
    private Logger log = LoggerFactory.getLogger(MQConsumerByKafka.class);
    private Map<String, ConcurrentMessageListenerContainer> mapTopic2ListenerContainer = new ConcurrentHashMap();
    private DefaultKafkaConsumerFactory consumerFactory = null;

    public MQConsumerByKafka() {
    }

    public void start() {
        if (this.properties != null && "false".equals(this.properties.getProperty("enabled"))) {
            return;
        }
        if (this.consumerFactory == null) {

            this.properties.put("enable.auto.commit", "false");
            this.properties.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
            this.properties.put("value.deserializer", "org.apache.kafka.common.serialization.BytesDeserializer");
            this.consumerFactory = new DefaultKafkaConsumerFactory(this.properties);
            if (this.listTopic2Handler != null && !this.listTopic2Handler.isEmpty()) {
                Iterator var1 = this.listTopic2Handler.iterator();

                while (var1.hasNext()) {
                    MQConsumerInfo mqConsumerInfo = (MQConsumerInfo) var1.next();
                    String tmpKey = this.getConsumerKey(mqConsumerInfo.getTopic(), mqConsumerInfo.getTag());
                    this.mapTopic2Concurrency.put(tmpKey, mqConsumerInfo.getConcurrency());
                    this.subscribe(tmpKey, mqConsumerInfo.getBusinessService());
                }

            }
        }
    }

    public void shutdown() {
        if (this.properties != null && "false".equals(this.properties.getProperty("enabled"))) {
            return;
        }
        Iterator var1 = this.mapTopic2ListenerContainer.keySet().iterator();

        while (var1.hasNext()) {
            String key = (String) var1.next();
            this.unsubscribe(key);
        }

        this.consumerFactory = null;
    }

    public void subscribe(String topic, MQBusinessHandler mqHandler) {
        String final_topic = this.getFinalTopic(topic);
        //以下代码可以实现一个消费者消费一个partition分区,那么concurrency需要配置成1
        TopicPartitionInitialOffset partition = TopicPartitionInitialOffset.fromTPO(new TopicPartitionOffset(final_topic, 1));
        //ContainerProperties containerProperties = new ContainerProperties(partition);
        ContainerProperties containerProperties = new ContainerProperties(Pattern.compile(topic));

        containerProperties.setMessageListener(this);
        ConcurrentMessageListenerContainer messageListenerContainer =
                new ConcurrentMessageListenerContainer(this.consumerFactory, containerProperties);
        ContainerProperties properties = messageListenerContainer.getContainerProperties();
        //containerProperties.getAckMode();
        properties.setAckMode(ContainerProperties.AckMode.MANUAL_IMMEDIATE);
        //配置开几个消费者去消费分区里面的数据 例如3个分区 开3个线程当做3个消费者去消费
        //代码位置:org.springframework.kafka.listener.ConcurrentMessageListenerContainer.doStart
        Integer concurrency = (Integer) this.mapTopic2Concurrency.get(topic);
        if (concurrency != null) {
            messageListenerContainer.setConcurrency(concurrency);
        }
        messageListenerContainer.start();
        this.mapTopic2ListenerContainer.put(topic, messageListenerContainer);
        this.mapTopic2Handler.put(final_topic, mqHandler);
    }

    public void unsubscribe(String topic) {
        Iterator it = this.mapTopic2ListenerContainer.entrySet().iterator();

        while (it.hasNext()) {
            Map.Entry<String, ConcurrentMessageListenerContainer> entry = (Map.Entry) it.next();
            boolean removeflag = false;
            if (((String) entry.getKey()).indexOf("%") != -1 && ((String) entry.getKey()).startsWith(topic)) {
                removeflag = true;
            }

            if (((String) entry.getKey()).indexOf("%") == -1 && ((String) entry.getKey()).equals(topic)) {
                removeflag = true;
            }

            if (removeflag) {
                ((ConcurrentMessageListenerContainer) entry.getValue()).stop();
                it.remove();
                this.mapTopic2Handler.remove(this.getFinalTopic((String) entry.getKey()));
            }
        }

    }

    private String getConsumerKey(String topic, String tag) {
        return tag != null && tag.trim().length() > 0 ? topic + "%" + tag : topic;
    }

    private String getFinalTopic(String topic) {
        return topic.replace("%", "-");
    }

    public Properties getProperties() {
        return this.properties;
    }

    public void setProperties(Properties properties) {
        this.properties = properties;
    }

    public List<MQConsumerInfo> getListTopic2Handler() {
        return this.listTopic2Handler;
    }

    public void setListTopic2Handler(List<MQConsumerInfo> listTopic2Handler) {
        this.listTopic2Handler = listTopic2Handler;
    }

    public Map<String, Integer> getMapTopic2Concurrency() {
        return this.mapTopic2Concurrency;
    }

    public void setMapTopic2Concurrency(Map<String, Integer> mapTopic2Concurrency) {
        this.mapTopic2Concurrency = mapTopic2Concurrency;
    }

    public Map<String, MQBusinessHandler> getMapTopic2Handler() {
        return mapTopic2Handler;
    }

    public void setMapTopic2Handler(Map<String, MQBusinessHandler> mapTopic2Handler) {
        this.mapTopic2Handler = mapTopic2Handler;
    }

    @Override
    public void onMessage(ConsumerRecord<String, Bytes> consumerRecord, Acknowledgment acknowledgment) {
        MQBusinessHandler mqBusinessHandler = (MQBusinessHandler) this.mapTopic2Handler.get(consumerRecord.topic());
        if (mqBusinessHandler == null) {
            this.log.error("Received a message, but it's not subscribed by mqBusinessHandler.Topic:" + consumerRecord.topic() + ",MsgID:" + (String) consumerRecord.key());
        } else {
            try {
                Bytes value = (Bytes) consumerRecord.value();
                MQMessage<?> mqMessage = (MQMessage) SerializationUtils.deserialize(value.get());
                if (mqBusinessHandler.doBusiness(mqMessage)) {
                    acknowledgment.acknowledge();
                }

            } catch (Exception e) {
                e.printStackTrace();
            }
        }
    }
}
